package com.explorex.infra.puma.gateway.netty.common.ack;


import com.explorex.infra.puma.gateway.netty.common.BaseSocketSession;
import com.explorex.infra.puma.gateway.netty.common.WaitingAckMessage;
import com.explorex.infra.puma.gateway.netty.common.enums.QosEnum;
import io.netty.util.HashedWheelTimer;
import io.netty.util.Timeout;
import io.netty.util.Timer;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.concurrent.BasicThreadFactory;

import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author: JacceYang
 * @createTime: 2022/08/22 12:54
 * @description: 消息重试逻辑的类
 */
@Slf4j
public class AckMessageManager {

    private static final Map<String, Timeout> ackTimeTask = new ConcurrentHashMap<>();
    private static final Timer timer = new HashedWheelTimer(50, TimeUnit.MILLISECONDS, 2048);

    private static final ScheduledExecutorService scheduledExecutor = new ScheduledThreadPoolExecutor(1, new BasicThreadFactory.Builder().namingPattern(AckMessageManager.class.getSimpleName() + "-%d").build());

    static {
        scheduledExecutor.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                cleanSelf();
            }
        }, 30, 15, TimeUnit.SECONDS);
    }

    private static void cleanSelf() {
        Iterator<Map.Entry<String, Timeout>> iterator = ackTimeTask.entrySet().iterator();
        while (iterator.hasNext()) {
            Map.Entry<String, Timeout> item = iterator.next();
            if (item.getValue() != null) {
                SendMessageTask task = (SendMessageTask) item.getValue().task();
                // 消息在服务器保存10 min 了，依然没有被删除掉，则强制删除数据
                if (System.currentTimeMillis() - task.waitingAckMessage.getEnQueueTime() > 1000 * 60 * 10) {
                    item.getValue().cancel();
                    iterator.remove();
                }
            }
        }
    }

    /**
     * 发送成功后记录到定时器中
     *
     * @param socketSession
     * @param message
     */
    public static void sendMessage(BaseSocketSession socketSession, WaitingAckMessage message) {
        sendMessage(socketSession,message,1500,3, (short) QosEnum.LEAST_ONCE.getValue());
    }

    /**
     * 发送成功后记录到定时器中
     *
     * @param socketSession
     * @param message
     */
    public static void sendMessage(BaseSocketSession socketSession, WaitingAckMessage message, long delayTime, int retryTimes, short qos) {
        // 已经ack 过的消息，也不需要做定时统计了
        if (message.getStatus() == AckStatusEnum.ACK.getValue()) {
            return;
        }
        Timeout timeout = timer.newTimeout(new SendMessageTask(socketSession, message, delayTime, retryTimes, qos), delayTime, TimeUnit.MILLISECONDS);
        ackTimeTask.put(socketSession.channel().id().asLongText() + "@" + message.getSeq(), timeout);
    }

    /**
     * 收到回复了，立即取消定时任务
     *
     * @param socketSession
     * @param seqId
     */
    public static WaitingAckMessage ackMessage(BaseSocketSession socketSession, long seqId, String messageType) {
        Timeout removeTask = ackTimeTask.remove(socketSession.channel().id().asLongText() + "@" + seqId);
        if (removeTask != null) {
            SendMessageTask task = (SendMessageTask) removeTask.task();
            task.waitingAckMessage.setStatus(AckStatusEnum.ACK.getValue());
            removeTask.cancel();
            // 如果是5s 内有回复,说明网路没有那么阻塞了，重新开启重试ack 能力。2000ms 为重试一次回复的的间隔，这个值应该小于首次重试间隔较为合理
            if (System.currentTimeMillis() - task.waitingAckMessage.getEnQueueTime() <= 2000) {
                socketSession.setAckRetryFuse(false);
            }
            return task.waitingAckMessage;
        }
        return null;
    }
    public static void delete(BaseSocketSession socketSession, WaitingAckMessage message, AckStatusEnum status) {
        Timeout deleteTask = ackTimeTask.remove(socketSession.channel().id().asLongText() + "@" + message.getSeq());
        if (deleteTask == null) {
            return;
        }
        deleteTask.cancel();
        SendMessageTask task = (SendMessageTask) deleteTask.task();
        // 不可写或者重试次数过多了的情况，说明天通道压力较大，需要暂时关闭重试功能.
        socketSession.setAckRetryFuse(true);
    }
}
